package test.threads.queue.unstressed;

import java.util.concurrent.Executor;

import com.lmax.disruptor.EventFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import test.helpers.Config;
import test.threads.queue.common.BaseEnqueuer;
import test.threads.queue.common.IUsefulWorkEmulator;
import test.threads.queue.stressed.AbstractTask;
import test.threads.queue.stressed.cpu.LongValueMessage;
import test.threads.queue.unstressed.impl.CABSESDMaskedLongQueue;

/**
 * @author cheremin
 * @since 24.03.12,  14:52
 */
public class LongQueueTask extends AbstractTask<LongValueMessage> {
    private static final Log log = LogFactory.getLog( LongQueueTask.class );

    private final CABSESDMaskedLongQueue queue;

    public LongQueueTask( final Config config ) throws Exception {
        super( config );

        this.queue = new CABSESDMaskedLongQueue( bufferSize() );
    }


    @Override
    protected BaseEnqueuer _initialize( final Executor threadPool,
                                        final IUsefulWorkEmulator<LongValueMessage>[] emulators,
                                        final EventFactory<LongValueMessage> factory ) throws Exception {
        return new Enqueuer();
    }

    @Override
    protected long dequeueBatchEvents( final int eventsInBatch ) throws Exception {
        long sum = 0;
        for ( int i = 0; i < eventsInBatch; i++ ) {
            sum += queue.dequeue();
        }
        return sum;
    }

    @Override
    public int workersRequired() {
        return 1;
    }

    @Override
    public String toString() {
        return String.format(
                "%s[CABSESDMaskedLongQueue[%d]]",
                getClass().getSimpleName(),
                bufferSize()
        );
    }

    private class Enqueuer extends BaseEnqueuer<LongValueMessage> {
        private Enqueuer() {
            super( eventsInBatch() );
        }

        @Override
        protected void enqueueEventsBatch( final int eventsInBatch ) throws Exception {
            for ( int i = 0; i < eventsInBatch; i++ ) {
                queue.enqueue( i );
            }
        }
    }
}
